package com.zhm.demo.rockmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * @author TAO
 * @description: 批量消息消费者
 * @date 2021/12/9 6:48 下午
 */
@Slf4j
public class BatchConsumer {

    /*
    {"data":[{"id":"6","name":"test02","ts":"2024-12-21 10:17:34","created_date":"2024-12-21 10:17:37","del_flag":"0"}],"database":"zhm","es":1734747685000,"gtid":"","id":2,"isDdl":false,"mysqlType":{"id":"int(11)","name":"varchar(255)","ts":"timestamp","created_date":"datetime","del_flag":"tinyint(1) unsigned"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12,"ts":93,"created_date":93,"del_flag":-6},"table":"test","ts":1734747685786,"type":"INSERT"}
     */

    public static void main(String[] args) throws Exception {
        RockMqConfig.Item sit = RockMqConfig.sit();
        //1.创建消费者Consumer，制定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr(sit.getServerAddr());
        // 设置消费模式为集群模式，默认就是集群模式，这里显式设置
        consumer.setMessageModel(MessageModel.CLUSTERING);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    for (MessageExt msg : msgs) {
                        log.info("{} {} {}", msg.getTopic(), msg.getTags(), new String(msg.getBody()));
                    }
                    // 手动提交 ACK，表示消息消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    // 消费失败，稍后重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

        //5.启动消费者consumer
        consumer.start();

        System.out.println("consumer start");

        Thread.sleep(10000);

        //3.订阅主题Topic和Tag
        consumer.subscribe("example", "*");

        System.out.println("consumer subscribe");

        Thread.sleep(20*1000L);

        consumer.subscribe("xxx", "*");

        System.out.println("consumer subscribe2");


//        consumer.unsubscribe("example");

//        consumer.shutdown();
    }
}
